package com.guandata.cdc.source;

import com.guandata.cdc.model.source.AbstractSourceConfig;
import lombok.Getter;
import org.apache.flink.table.api.Expressions;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.expressions.Expression;

@Getter
public abstract class GuanDataSourceFactory<T extends AbstractSourceConfig> {
    protected final T config;

    protected Schema schema;

    public GuanDataSourceFactory(T config) {
        this.config = config;
    }

    public abstract Table registerSourceTable(StreamTableEnvironment tableEnv);

    public Table dropColumnsFrom(Table table) {
        var wannaDropColumns = config.dropColumns.stream()
            .map(Expressions::$)
            .toArray(Expression[]::new);

        if(wannaDropColumns.length > 0) {
            return table.dropColumns(wannaDropColumns);
        }

        return table;
    }
}
